package com.base.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

@Configuration
@Slf4j
public class ProducerRabbitmqConfig {


    @Resource
    private RabbitTemplate rabbitTemplate;


    /**
     * 定制化amqp模版      可根据需要定制多个
     * <p>
     * <p>
     * 此处为模版类定义 Jackson消息转换器
     * ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调   即消息发送到exchange  ack
     * ReturnCallback接口用于实现消息发送到RabbitMQ 交换器，但无相应队列与交换器绑定时的回调  即消息发送不到任何一个队列中  ack
     *
     * @return the amqp template
     */
    @Bean
    public AmqpTemplate amqpTemplate() {
        //使用jackson 消息转换器
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        rabbitTemplate.setEncoding("UTF-8");
        //开启returncallback     yml 需要 配置    publisher-returns: true
        rabbitTemplate.setMandatory(true);

        /*
        rabbitTemplate.setConfirmCallback()
        消息发送到 Broker 后触发回调，确认消息是否到达 Broker 服务器，也就是只确认是否正确到达 Exchange 中

        rabbitTemplate.setReturnCallback()
        通过实现 ReturnCallback 接口，启动消息失败返回，比如路由不到队列时触发回调

        需要
        publisher-confirm-type: correlated
        publisher-returns: true 才会生效
        */
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("消息：{} 发送失败, 应答码：{} 原因：{} 交换机: {}  路由键: {}", replyCode, replyText, exchange, routingKey);
        });
        //消息确认模式  yml 需要配置   publisher-returns: true
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("消息发送到exchange成功,id: {}", correlationData);
            } else {
                log.info("消息发送到exchange失败,原因: {}", cause);
            }
        });


        return rabbitTemplate;
    }


    /**
     * 声明Direct交换机 支持持久化.
     *
     * @return the exchange
     * durable: 消息队列持久化
     */
    @Bean("topicExchange")
    public Exchange topicExchange() {
        return ExchangeBuilder.topicExchange("TOPIC_EXCHANGE").durable(true).build();
    }

    /**
     * 声明一个队列 支持持久化.
     *   # 匹配一个或多个，* 匹配一个，一般使用#号匹配多个，*号用的比较少。
     * @return the queue
     */
    @Bean("topicQueue1")
    public Queue topicQueue1() {
        return QueueBuilder.durable("TOPIC_QUEUE1").build();
    }

    /**
     * 声明一个队列 支持持久化.
     *
     * @return the queue
     */
    @Bean("topicQueue2")
    public Queue topicQueue2() {
        return QueueBuilder.durable("TOPIC_QUEUE2").build();
    }

    @Bean
    //# 匹配一个或多个，* 匹配一个，一般使用#号匹配多个，*号用的比较少
    public Binding topicBinding1(@Qualifier("topicQueue1") Queue queue, @Qualifier("topicExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("TOPIC.INFO.*").noargs(); // *配置一个路由
    }
    @Bean
    public Binding topicBinding2(@Qualifier("topicQueue2") Queue queue, @Qualifier("topicExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("TOPIC.#").noargs(); // 配置#无条件路由
    }
    /* ----------------------------------------------------------------------------Direct exchange test--------------------------------------------------------------------------- */

    /**
     * 声明Direct交换机 支持持久化.
     *
     * @return the exchange
     * durable: 消息队列持久化
     */
    @Bean("directExchange")
    public Exchange directExchange() {
        return ExchangeBuilder.directExchange("DIRECT_EXCHANGE").durable(true).build();
    }

    /**
     * 声明一个队列 支持持久化.
     *
     * @return the queue
     */
    @Bean("directQueue")
    public Queue directQueue() {
        return QueueBuilder.durable("DIRECT_QUEUE").build();
    }

    /**
     * 通过绑定键 将指定队列绑定到一个指定的交换机 .
     *
     * @param queue    the queue
     * @param exchange the exchange
     * @return the binding
     */
    @Bean
    public Binding directBinding(@Qualifier("directQueue") Queue queue, @Qualifier("directExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("DIRECT_ROUTING_KEY").noargs();
    }

    /* ----------------------------------------------------------------------------Fanout exchange test--------------------------------------------------------------------------- */

    /**
     * 声明 fanout 交换机.
     *
     * @return the exchange
     */
    @Bean("fanoutExchange")
    public FanoutExchange fanoutExchange() {
        return (FanoutExchange) ExchangeBuilder.fanoutExchange("FANOUT_EXCHANGE").durable(true).build();
    }

    /**
     * Fanout queue A.
     *
     * @return the queue
     */
    @Bean("fanoutQueueA")
    public Queue fanoutQueueA() {
        return QueueBuilder.durable("FANOUT_QUEUE_A").build();
    }

    /**
     * Fanout queue B .
     *
     * @return the queue
     */
    @Bean("fanoutQueueB")
    public Queue fanoutQueueB() {
        return QueueBuilder.durable("FANOUT_QUEUE_B").build();
    }

    /**
     * 绑定队列A 到Fanout 交换机.
     *
     * @param queue          the queue
     * @param fanoutExchange the fanout exchange
     * @return the binding
     */
    @Bean
    public Binding bindingA(@Qualifier("fanoutQueueA") Queue queue, @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }

    /**
     * 绑定队列B 到Fanout 交换机.
     *
     * @param queue          the queue
     * @param fanoutExchange the fanout exchange
     * @return the binding
     */
    @Bean
    public Binding bindingB(@Qualifier("fanoutQueueB") Queue queue, @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }

    /**
     * 延时队列
     * 消息被拒绝（basic.reject/ basic.nack）并且requeue=false
     * 消息TTL过期（参考：RabbitMQ之TTL（Time-To-Live 过期时间））
     * 队列达到最大长度
     *
     * @return
     */
    @Bean("delayQueue")
    public Queue delayQueue() {
        //生产者发布的消息路由到该延迟队列，30秒后消息成为死信并被重新发送到死信交换器DLX中；消费者订阅死信队列DLQ，消费消息。
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-message-ttl", 30 * 1000); // Message TTL = 30s 聲明30秒延遲隊列
        //arguments.put("x-max-length", 1000000); // Max length = 1million
        // 队列中的消息变成死信后，进入死信交换机
        arguments.put("x-dead-letter-exchange", "DIRECT_EXCHANGE");  //DLX
        arguments.put("x-dead-letter-routing-key", "DIRECT_ROUTING_KEY"); //DLK
        //创建延时队列DELAY_QUEUE，如果超时则将消息转发到DIRECT_EXCHANGE DLX 中
        Queue queue = new Queue("DELAY_QUEUE", true, false, false, arguments);
        return queue;
    }

    @Bean("delayExchange")
    public Exchange delayExchange() {
        return ExchangeBuilder.directExchange("DELAY_EXCHANGE").durable(true).build();
    }

    @Bean
    public Binding delayBinding(@Qualifier("delayQueue") Queue queue, @Qualifier("delayExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("DELAY_ROUTING_KEY").noargs(); // 配置#无条件路由
    }

    /**
     * 超时队列
     * 消息被拒绝（basic.reject/ basic.nack）并且requeue=false
     * 消息TTL过期（参考：RabbitMQ之TTL（Time-To-Live 过期时间））
     * 队列达到最大长度
     *
     * @return
     */
    @Bean("deadLetterQueue")
    public Queue deadLetterQueue() {
        Map<String, Object> arguments = new HashMap<>();
        // 队列中的消息变成死信后，进入死信交换机
        arguments.put("x-dead-letter-exchange", "DIRECT_EXCHANGE");  //DLX
        arguments.put("x-dead-letter-routing-key", "DIRECT_ROUTING_KEY"); //DLK
        //创建队列DEAD_LETTER_QUEUE，如果消息超时则将消息转发到DIRECT_EXCHANGE DLX 中
        Queue queue = new Queue("DEAD_LETTER_QUEUE", true, false, false, arguments);
        return queue;
    }

    @Bean
    public Binding deadLetterBinding(@Qualifier("deadLetterQueue") Queue queue, @Qualifier("directExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("DIRECT_ROUTING_KEY2").noargs();
    }

    /**
     * 优先级队列
     * 优先级队列，顾名思义，具有更高优先级的队列具有较高的优先权，优先级高的消息具备优先被消费的特权。
     * 在消费端速度大于生产端速度，且broker中没有消息堆积的话，对发送的消息设置优先级也没什么实际意义，因为发送端刚发送完一条消息就被消费端消费了，那么就相当于broker至多只有一条消息，那么对于单条消息来说优先级是没有什么意义的。
     * @return
     */
//    @Bean("priorityQueue")
//    public Queue priorityQueue() {
//        Map<String, Object> arguments = new HashMap<>();
//        arguments.put("x-max-priority", 10); //队列的属性参数 有10个优先级别
//        Queue queue = new Queue("PRIORITY_QUEUE", true, false, false, arguments);
//        return queue;
//    }
//
//    @Bean
//    public Binding priorityBinding(@Qualifier("priorityQueue") Queue queue, @Qualifier("directExchange") Exchange exchange) {
//        return BindingBuilder.bind(queue).to(exchange).with("DIRECT_ROUTING_KEY").noargs();
//    }


//    优先级队列：
//    1.只有当消费者不足，不能及时进行消费的前提下，优先级队列才能发挥作用。
//    2.队列、消息都要设置优先级，优先级队列才可生效，以较小值为准。
//    3.队列优先级只能声明一次，不可改变。
//    4.优先级队列在内存、磁盘、CPU都会有消耗，不建议创建大量的优先级队列。(劳民伤财的事要少干)
//    5.在消费端速度大于生产端速度，且broker中没有消息堆积的话，对发送的消息设置优先级也没什么实际意义，因为发送端刚发送完一条消息就被消费端消费了，那么就相当于broker至多只有一条消息，那么对于单条消息来说优先级是没有什么意义的。
//    6.有指定优先级的消息会将优先级以0对待。 对于超过优先级队列所定最大优先级的消息，优先级以最大优先级对待。对于相同优先级的消息，后进的排在前面
}

